package com.zxq.mall.realtime.app

import java.sql.Date
import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}

import com.alibaba.fastjson.JSON
import com.zxq.gmall.realtime.util.MyEsUtils
import com.zxq.mall.realtime.bean.{DauInfo, PageLog}
import com.zxq.mall.realtime.util.{MyBeanUtils, MyKafkaUtils, MyRedisUtils, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object DwDDauApp {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("dwd_dau_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf,Seconds(5))

    val topic = "DWD_PAGE_LOG_TOPIC"
    val groupId = "dwd_dau_group"

    val offsets : Map[TopicPartition,Long]= OffsetManagerUtil.getOffset(topic,groupId)

    var kafkaDStream : DStream[ConsumerRecord[String, String]] = null
    if(offsets != null && offsets.nonEmpty){
        kafkaDStream = MyKafkaUtils.getKafkaDStream(topic,ssc,offsets,groupId)
    }else{
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc,topic,groupId)
    }
    var offsetRanges : Array[OffsetRange] = null
    kafkaDStream = kafkaDStream.transform(
      rdd => {
        rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    val pageLogDStream = kafkaDStream.map(
      record => {
        val jsonStr = record.value()
        val pageLog = JSON.parseObject(jsonStr, classOf[PageLog])
        pageLog
      }
    )
    val filterDStreamByLastPage: DStream[PageLog] = pageLogDStream.filter(
      pageLog => {
        pageLog.last_page_id == null
      }
    )

    val pageLogFilterDStream: DStream[PageLog] = filterDStreamByLastPage.mapPartitions(
      pageLogIter => {
        val jedis: Jedis = MyRedisUtils.getJedisClient
        val filterList: ListBuffer[PageLog] =
          ListBuffer[PageLog]()
        val pageLogList: List[PageLog] = pageLogIter.toList
        println("过滤前 : " + pageLogList.size)
        for (pageLog <- pageLogList) {
          val sdf = new SimpleDateFormat("yyyy-MM-dd")
          val dateStr: String = sdf.format(new Date(pageLog.ts))
          val dauKey: String = s"DAU:$dateStr"
          val ifNew: Long = jedis.sadd(dauKey, pageLog.mid)

          jedis.expire(dauKey,3600 * 24)
          if (ifNew == 1L) {
            filterList.append(pageLog)
          }
        }
        jedis.close()
        println("过滤后: " + filterList.size)
        filterList.toIterator
      }
    )
    val dauInfoDStream : DStream[DauInfo] = pageLogFilterDStream.mapPartitions(
      pageLogIter => {
        val jedis : Jedis = MyRedisUtils.getJedisClient
        val dauInfoList = ListBuffer[DauInfo]()
        for (pageLog <- pageLogIter) {
          val dimUserKey = s"DIM:USER_INFO:${pageLog.user_id}"
          val userInfoJson = jedis.get(dimUserKey)
          val userInfoJsonObj = JSON.parseObject(userInfoJson)
          val birthday = userInfoJsonObj.getString("birthday")
          val gender = userInfoJsonObj.getString("gender")//性别
          var age : String = null
          if( birthday != null){
            val birthdayDate = LocalDate.parse(birthday)
            val nowDate = LocalDate.now()
            val period : Period = Period.between(birthdayDate,nowDate)
            val years : Int = period.getYears
            age = years.toString

          }
          val dauInfo = new DauInfo()
          MyBeanUtils.copyProperties(pageLog,dauInfo)
          dauInfo.user_gender = gender
          dauInfo.user_age = age
          val provinceKey : String = s"DIM:BASE_PROVINCE:${pageLog.province_id}"
          val provinceJson = jedis.get(provinceKey)
          if(provinceJson != null && provinceJson.nonEmpty){
            val provinceJsonObj = JSON.parseObject(provinceJson)
            dauInfo.province_name = provinceJsonObj.getString("name")
            dauInfo.province_area_code = provinceJsonObj.getString("area_code")
            dauInfo.province_3166_2 = provinceJsonObj.getString("iso_3166_2")
            dauInfo.province_iso_code = provinceJsonObj.getString("iso_code")

          }
          val dateFormat = new SimpleDateFormat("yyyy-MM-dd hh")
          val dtDate = new Date(dauInfo.ts)
          val dtHr = dateFormat.format(dtDate)
          val dtHrArray = dtHr.split(" ")
          dauInfo.dt = dtHrArray(0)
          dauInfo.hr = dtHrArray(1)

          dauInfoList.append(dauInfo)
        }
        jedis.close()
        dauInfoList.toIterator
      }
    )
    dauInfoDStream.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          dauInfoIter => {

            val dauInfos: List[(String, DauInfo)] =
            dauInfoIter.toList.map(dauInfo => (dauInfo.mid,dauInfo))
            if(dauInfos.size > 0 ){

              val dauInfoT: (String, DauInfo) = dauInfos(0)
              val dt = dauInfoT._2.dt

              MyEsUtils.bulkSaveIdempotent(dauInfos,s"gmall_dau_info_$dt")
            }
          }
        )

        OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)
      }
    )
    ssc.start()
    ssc.awaitTermination()
  }

}
